package processor

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/processing/data"
)

// ProcessorOptions 表示处理器选项
type ProcessorOptions struct {
	// MaxBatchSize 是最大批处理大小
	MaxBatchSize int
	// MaxBatchWait 是最大批处理等待时间
	MaxBatchWait time.Duration
	// Concurrency 是并发度
	Concurrency int
	// BufferSize 是缓冲区大小
	BufferSize int
}

// DefaultProcessorOptions 返回默认处理器选项
func DefaultProcessorOptions() *ProcessorOptions {
	return &ProcessorOptions{
		MaxBatchSize: 100,
		MaxBatchWait: time.Second,
		Concurrency:  4,
		BufferSize:   1000,
	}
}

// ProcessorFunc 表示处理函数
type ProcessorFunc func(ctx context.Context, record *data.DataRecord) (*data.DataRecord, error)

// BatchProcessorFunc 表示批处理函数
type BatchProcessorFunc func(ctx context.Context, batch *data.DataBatch) (*data.DataBatch, error)

// Processor 表示处理器
type Processor struct {
	// name 是处理器名称
	name string
	// processorFunc 是处理函数
	processorFunc ProcessorFunc
	// batchProcessorFunc 是批处理函数
	batchProcessorFunc BatchProcessorFunc
	// options 是处理器选项
	options *ProcessorOptions
	// inputCh 是输入通道
	inputCh chan *data.DataRecord
	// outputCh 是输出通道
	outputCh chan *data.DataRecord
	// errorCh 是错误通道
	errorCh chan error
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
	// wg 是等待组
	wg sync.WaitGroup
}

// NewProcessor 创建一个新的处理器
func NewProcessor(name string, processorFunc ProcessorFunc, options *ProcessorOptions) *Processor {
	if options == nil {
		options = DefaultProcessorOptions()
	}

	ctx, cancel := context.WithCancel(context.Background())

	return &Processor{
		name:          name,
		processorFunc: processorFunc,
		options:       options,
		inputCh:       make(chan *data.DataRecord, options.BufferSize),
		outputCh:      make(chan *data.DataRecord, options.BufferSize),
		errorCh:       make(chan error, options.BufferSize),
		ctx:           ctx,
		cancel:        cancel,
	}
}

// NewBatchProcessor 创建一个新的批处理器
func NewBatchProcessor(name string, batchProcessorFunc BatchProcessorFunc, options *ProcessorOptions) *Processor {
	if options == nil {
		options = DefaultProcessorOptions()
	}

	ctx, cancel := context.WithCancel(context.Background())

	return &Processor{
		name:              name,
		batchProcessorFunc: batchProcessorFunc,
		options:           options,
		inputCh:           make(chan *data.DataRecord, options.BufferSize),
		outputCh:          make(chan *data.DataRecord, options.BufferSize),
		errorCh:           make(chan error, options.BufferSize),
		ctx:               ctx,
		cancel:            cancel,
	}
}

// Start 启动处理器
func (p *Processor) Start() error {
	// 启动工作协程
	for i := 0; i < p.options.Concurrency; i++ {
		p.wg.Add(1)
		if p.batchProcessorFunc != nil {
			go p.batchProcessLoop(i)
		} else {
			go p.processLoop(i)
		}
	}

	return nil
}

// Stop 停止处理器
func (p *Processor) Stop() error {
	// 取消上下文
	p.cancel()

	// 等待工作协程结束
	p.wg.Wait()

	// 关闭通道
	close(p.inputCh)
	close(p.outputCh)
	close(p.errorCh)

	return nil
}

// Process 处理记录
func (p *Processor) Process(ctx context.Context, record *data.DataRecord) (*data.DataRecord, error) {
	if p.processorFunc != nil {
		return p.processorFunc(ctx, record)
	}
	return record, nil
}

// ProcessBatch 处理批次
func (p *Processor) ProcessBatch(ctx context.Context, batch *data.DataBatch) (*data.DataBatch, error) {
	if p.batchProcessorFunc != nil {
		return p.batchProcessorFunc(ctx, batch)
	}
	return batch, nil
}

// Input 获取输入通道
func (p *Processor) Input() chan<- *data.DataRecord {
	return p.inputCh
}

// Output 获取输出通道
func (p *Processor) Output() <-chan *data.DataRecord {
	return p.outputCh
}

// Errors 获取错误通道
func (p *Processor) Errors() <-chan error {
	return p.errorCh
}

// processLoop 处理循环
func (p *Processor) processLoop(workerID int) {
	defer p.wg.Done()

	for {
		select {
		case <-p.ctx.Done():
			return
		case record, ok := <-p.inputCh:
			if !ok {
				return
			}

			// 处理记录
			result, err := p.Process(p.ctx, record)
			if err != nil {
				select {
				case p.errorCh <- fmt.Errorf("processor %s worker %d: %w", p.name, workerID, err):
				default:
					// 错误通道已满，记录错误
					fmt.Printf("Processor %s worker %d: error channel full, dropping error: %v\n", p.name, workerID, err)
				}
				continue
			}

			// 发送结果
			if result != nil {
				select {
				case p.outputCh <- result:
				default:
					// 输出通道已满，记录错误
					select {
					case p.errorCh <- fmt.Errorf("processor %s worker %d: output channel full", p.name, workerID):
					default:
						fmt.Printf("Processor %s worker %d: error channel full, dropping error\n", p.name, workerID)
					}
				}
			}
		}
	}
}

// batchProcessLoop 批处理循环
func (p *Processor) batchProcessLoop(workerID int) {
	defer p.wg.Done()

	batch := data.NewDataBatch()
	timer := time.NewTimer(p.options.MaxBatchWait)
	defer timer.Stop()

	for {
		select {
		case <-p.ctx.Done():
			// 处理剩余的批次
			if batch.Count() > 0 {
				p.processBatch(batch, workerID)
			}
			return
		case record, ok := <-p.inputCh:
			if !ok {
				// 处理剩余的批次
				if batch.Count() > 0 {
					p.processBatch(batch, workerID)
				}
				return
			}

			// 添加记录到批次
			batch.AddRecord(record)

			// 如果批次已满，处理批次
			if batch.Count() >= p.options.MaxBatchSize {
				p.processBatch(batch, workerID)
				batch = data.NewDataBatch()
				timer.Reset(p.options.MaxBatchWait)
			}
		case <-timer.C:
			// 如果批次不为空，处理批次
			if batch.Count() > 0 {
				p.processBatch(batch, workerID)
				batch = data.NewDataBatch()
			}
			timer.Reset(p.options.MaxBatchWait)
		}
	}
}

// processBatch 处理批次
func (p *Processor) processBatch(batch *data.DataBatch, workerID int) {
	// 处理批次
	result, err := p.ProcessBatch(p.ctx, batch)
	if err != nil {
		select {
		case p.errorCh <- fmt.Errorf("processor %s worker %d: %w", p.name, workerID, err):
		default:
			// 错误通道已满，记录错误
			fmt.Printf("Processor %s worker %d: error channel full, dropping error: %v\n", p.name, workerID, err)
		}
		return
	}

	// 发送结果
	if result != nil {
		for _, record := range result.Records {
			select {
			case p.outputCh <- record:
			default:
				// 输出通道已满，记录错误
				select {
				case p.errorCh <- fmt.Errorf("processor %s worker %d: output channel full", p.name, workerID):
				default:
					fmt.Printf("Processor %s worker %d: error channel full, dropping error\n", p.name, workerID)
				}
				break
			}
		}
	}
}

// ProcessorChain 表示处理器链
type ProcessorChain struct {
	// processors 是处理器列表
	processors []*Processor
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
	// wg 是等待组
	wg sync.WaitGroup
}

// NewProcessorChain 创建一个新的处理器链
func NewProcessorChain(processors ...*Processor) *ProcessorChain {
	ctx, cancel := context.WithCancel(context.Background())

	return &ProcessorChain{
		processors: processors,
		ctx:        ctx,
		cancel:     cancel,
	}
}

// Start 启动处理器链
func (c *ProcessorChain) Start() error {
	// 启动所有处理器
	for _, processor := range c.processors {
		err := processor.Start()
		if err != nil {
			return fmt.Errorf("failed to start processor: %w", err)
		}
	}

	// 连接处理器
	for i := 0; i < len(c.processors)-1; i++ {
		current := c.processors[i]
		next := c.processors[i+1]

		// 启动转发协程
		c.wg.Add(1)
		go func(from *Processor, to *Processor) {
			defer c.wg.Done()
			c.forwardRecords(from, to)
		}(current, next)

		// 启动错误处理协程
		c.wg.Add(1)
		go func(processor *Processor) {
			defer c.wg.Done()
			c.handleErrors(processor)
		}(current)
	}

	// 处理最后一个处理器的错误
	if len(c.processors) > 0 {
		last := c.processors[len(c.processors)-1]
		c.wg.Add(1)
		go func(processor *Processor) {
			defer c.wg.Done()
			c.handleErrors(processor)
		}(last)
	}

	return nil
}

// Stop 停止处理器链
func (c *ProcessorChain) Stop() error {
	// 取消上下文
	c.cancel()

	// 停止所有处理器
	for _, processor := range c.processors {
		err := processor.Stop()
		if err != nil {
			return fmt.Errorf("failed to stop processor: %w", err)
		}
	}

	// 等待所有协程结束
	c.wg.Wait()

	return nil
}

// Input 获取输入通道
func (c *ProcessorChain) Input() chan<- *data.DataRecord {
	if len(c.processors) > 0 {
		return c.processors[0].Input()
	}
	return nil
}

// Output 获取输出通道
func (c *ProcessorChain) Output() <-chan *data.DataRecord {
	if len(c.processors) > 0 {
		return c.processors[len(c.processors)-1].Output()
	}
	return nil
}

// forwardRecords 转发记录
func (c *ProcessorChain) forwardRecords(from, to *Processor) {
	for {
		select {
		case <-c.ctx.Done():
			return
		case record, ok := <-from.Output():
			if !ok {
				return
			}

			// 转发记录
			select {
			case to.Input() <- record:
			case <-c.ctx.Done():
				return
			}
		}
	}
}

// handleErrors 处理错误
func (c *ProcessorChain) handleErrors(processor *Processor) {
	for {
		select {
		case <-c.ctx.Done():
			return
		case err, ok := <-processor.Errors():
			if !ok {
				return
			}

			// 处理错误
			fmt.Printf("Processor chain error: %v\n", err)
		}
	}
}

// FilterProcessor 表示过滤处理器
func FilterProcessor(predicate func(*data.DataRecord) bool) ProcessorFunc {
	return func(ctx context.Context, record *data.DataRecord) (*data.DataRecord, error) {
		if predicate(record) {
			return record, nil
		}
		return nil, nil
	}
}

// MapProcessor 表示映射处理器
func MapProcessor(mapper func(*data.DataRecord) (*data.DataRecord, error)) ProcessorFunc {
	return func(ctx context.Context, record *data.DataRecord) (*data.DataRecord, error) {
		return mapper(record)
	}
}

// FlatMapProcessor 表示扁平映射处理器
func FlatMapProcessor(mapper func(*data.DataRecord) ([]*data.DataRecord, error)) BatchProcessorFunc {
	return func(ctx context.Context, batch *data.DataBatch) (*data.DataBatch, error) {
		result := data.NewDataBatch()

		for _, record := range batch.Records {
			records, err := mapper(record)
			if err != nil {
				return nil, err
			}

			for _, r := range records {
				result.AddRecord(r)
			}
		}

		return result, nil
	}
}

// AggregateProcessor 表示聚合处理器
func AggregateProcessor(initialState interface{}, aggregator func(state interface{}, record *data.DataRecord) interface{}, resultMapper func(state interface{}) (*data.DataRecord, error)) BatchProcessorFunc {
	return func(ctx context.Context, batch *data.DataBatch) (*data.DataBatch, error) {
		state := initialState

		// 聚合记录
		for _, record := range batch.Records {
			state = aggregator(state, record)
		}

		// 映射结果
		result, err := resultMapper(state)
		if err != nil {
			return nil, err
		}

		// 创建结果批次
		resultBatch := data.NewDataBatch()
		if result != nil {
			resultBatch.AddRecord(result)
		}

		return resultBatch, nil
	}
}

// WindowProcessor 表示窗口处理器
type WindowProcessor struct {
	// windowSize 是窗口大小
	windowSize time.Duration
	// windowSlide 是窗口滑动
	windowSlide time.Duration
	// aggregator 是聚合函数
	aggregator func(records []*data.DataRecord) ([]*data.DataRecord, error)
	// windows 是窗口映射表
	windows map[int64][]*data.DataRecord
	// mu 是互斥锁
	mu sync.Mutex
	// lastTriggerTime 是上次触发时间
	lastTriggerTime time.Time
}

// NewWindowProcessor 创建一个新的窗口处理器
func NewWindowProcessor(windowSize, windowSlide time.Duration, aggregator func([]*data.DataRecord) ([]*data.DataRecord, error)) *WindowProcessor {
	return &WindowProcessor{
		windowSize:      windowSize,
		windowSlide:     windowSlide,
		aggregator:      aggregator,
		windows:         make(map[int64][]*data.DataRecord),
		lastTriggerTime: time.Now(),
	}
}

// Process 处理记录
func (p *WindowProcessor) Process(ctx context.Context, record *data.DataRecord) (*data.DataRecord, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// 计算记录所属的窗口
	timestamp := record.Timestamp.UnixNano()
	windowStart := timestamp - (timestamp % p.windowSlide.Nanoseconds())

	// 添加记录到窗口
	p.windows[windowStart] = append(p.windows[windowStart], record)

	// 检查是否需要触发窗口
	now := time.Now()
	if now.Sub(p.lastTriggerTime) >= p.windowSlide {
		p.lastTriggerTime = now
		return p.triggerWindows()
	}

	return nil, nil
}

// triggerWindows 触发窗口
func (p *WindowProcessor) triggerWindows() (*data.DataRecord, error) {
	// 计算过期窗口
	now := time.Now().UnixNano()
	expiredWindows := make([]int64, 0)
	for windowStart := range p.windows {
		if windowStart+p.windowSize.Nanoseconds() <= now {
			expiredWindows = append(expiredWindows, windowStart)
		}
	}

	// 处理过期窗口
	if len(expiredWindows) > 0 {
		// 收集所有过期窗口的记录
		allRecords := make([]*data.DataRecord, 0)
		for _, windowStart := range expiredWindows {
			allRecords = append(allRecords, p.windows[windowStart]...)
			delete(p.windows, windowStart)
		}

		// 聚合记录
		results, err := p.aggregator(allRecords)
		if err != nil {
			return nil, err
		}

		// 返回第一个结果
		if len(results) > 0 {
			return results[0], nil
		}
	}

	return nil, nil
}

// ProcessBatch 处理批次
func (p *WindowProcessor) ProcessBatch(ctx context.Context, batch *data.DataBatch) (*data.DataBatch, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// 添加所有记录到窗口
	for _, record := range batch.Records {
		// 计算记录所属的窗口
		timestamp := record.Timestamp.UnixNano()
		windowStart := timestamp - (timestamp % p.windowSlide.Nanoseconds())

		// 添加记录到窗口
		p.windows[windowStart] = append(p.windows[windowStart], record)
	}

	// 检查是否需要触发窗口
	now := time.Now()
	if now.Sub(p.lastTriggerTime) >= p.windowSlide {
		p.lastTriggerTime = now

		// 计算过期窗口
		nowNano := now.UnixNano()
		expiredWindows := make([]int64, 0)
		for windowStart := range p.windows {
			if windowStart+p.windowSize.Nanoseconds() <= nowNano {
				expiredWindows = append(expiredWindows, windowStart)
			}
		}

		// 处理过期窗口
		if len(expiredWindows) > 0 {
			// 收集所有过期窗口的记录
			allRecords := make([]*data.DataRecord, 0)
			for _, windowStart := range expiredWindows {
				allRecords = append(allRecords, p.windows[windowStart]...)
				delete(p.windows, windowStart)
			}

			// 聚合记录
			results, err := p.aggregator(allRecords)
			if err != nil {
				return nil, err
			}

			// 创建结果批次
			resultBatch := data.NewDataBatch()
			for _, result := range results {
				resultBatch.AddRecord(result)
			}

			return resultBatch, nil
		}
	}

	return data.NewDataBatch(), nil
}
